package com.atguigu.flink.chapter06;

import com.atguigu.flink.bean.UserBehavior;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.HashSet;
import java.util.Set;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/10 14:20
 */
public class Flink05_UV {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.读取数据
        SingleOutputStreamOperator<UserBehavior> userBehaviorDS = env
                .readTextFile("F:\\atguigu\\01_course\\code\\flink210323\\input\\UserBehavior.csv")
                .map(r -> {
                    String[] datas = r.split(",");
                    return new UserBehavior(
                            Long.parseLong(datas[0]),
                            Long.parseLong(datas[1]),
                            Integer.parseInt(datas[2]),
                            datas[3],
                            Long.parseLong(datas[4])
                    );
                });

        // 2.处理数据
        // 2.1 优先考虑过滤: 只保留 pv 的数据
        SingleOutputStreamOperator<UserBehavior> pvDS = userBehaviorDS.filter(r -> "pv".equals(r.getBehavior()));
        // TODO 使用Set去重，把用户ID存进去，最后 获取 Set的长度，就是 UV值
        // 2.2 使用map，丢弃没用的字段，减少数据量
        SingleOutputStreamOperator<Long> userIdDS = pvDS.map(new MapFunction<UserBehavior, Long>() {
            @Override
            public Long map(UserBehavior value) throws Exception {
                return value.getUserId();
            }
        });
        // 2.3 使用process 实现
        userIdDS
                .process(new ProcessFunction<Long, Integer>() {
                    // 定义一个Set结构，存 用户ID
                    private Set<Long> uvSet = new HashSet<Long>();

                    @Override
                    public void processElement(Long value, Context ctx, Collector<Integer> out) throws Exception {
                        uvSet.add(value);
                        out.collect(uvSet.size());
                    }
                }).setParallelism(1)
                .print().setParallelism(1);


        env.execute();

        /**
         * 另一种思路，不用 指定并行度1
         *      =》 转换成  ("pv",userID)  ==> keyby(pv) ==> process里面将 userID放进 Set
         */


    }
}
